/*
 * Copyright 2012 The Netty Project
 *
 * The Netty Project licenses this file to you under the Apache License,
 * version 2.0 (the "License"); you may not use this file except in compliance
 * with the License. You may obtain a copy of the License at:
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations
 * under the License.
 */
package io.netty.example.factorial;

import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

import java.math.BigInteger;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * Handler for a client-side channel.  This handler maintains stateful
 * information which is specific to a certain channel using member variables.
 * Therefore, an instance of this handler can cover only one channel.  You have
 * to create a new handler instance whenever you create a new channel and insert
 * this handler to avoid a race condition.
 *
 * 客户端处理器
 * 这个处理器, 维护的是一个有状态的信息 , 使用一个变量(在encode中说明) ,来确定的使用一个channel
 * 那么 , 当前处理器,只能处理这个特定的channel
 *
 * 重点:  ↓
 * 那么 , 当创建一个新的channel,并且想把当前handler放入channel时,
 * 必填创建一个新的实例来避免条件竞争
 *
 * 为什么会存在条件竞争 :
 * 当前实例对象中 ,保存了当前操作的数据
 * 如果其他的channel中也使用了这个实例对象
 *
 * 那么这个数据(接收的数据,next加的数据)将会产生并发.数据就不准确了
 *
 */
public class FactorialClientHandler extends SimpleChannelInboundHandler<BigInteger> {
    /**
     *
     */
    //当前channel的上下文
    private ChannelHandlerContext ctx;
    //接收的数据
    private int receivedMessages;
    //起始传到 server 的数据
    private int next = 10000000;

    /**
     * 一个线程安全的阻塞队列
     * 放入最终的答案
     * TODO : 这里为什么要设置成一个 线程安全的阻塞队列 ?
     */
    final BlockingQueue<BigInteger> answer = new LinkedBlockingQueue<BigInteger>();

    // 获取最终于阶乘结果
    public BigInteger getFactorial() {
        // 这里定义 了一个死循环,如果拿不到数据,则一直陷入死循环.
        // 直到有数据返回
        boolean interrupted = false;
        try {
            for (;;) {
                try {
                    return answer.take();
                } catch (InterruptedException ignore) {
                    interrupted = true;
                }
            }
        } finally {
            if (interrupted) {
                Thread.currentThread().interrupt();
            }
        }
    }

    /**
     * channel建立 通知
     * @param ctx
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        //获取当前channel的上下文
        this.ctx = ctx;
        //channel建立,链接成功 , 可以发送数据了.
        sendNumbers();
    }

    /**
     * 读取服务端传来的数据 ,
     *
     * msg : 服务端数据流ButeBuf ,到这里 ,已经被BigIntegerDecoder解析为BigInteger
     */
    @Override
    public void channelRead0(ChannelHandlerContext ctx, final BigInteger msg) {
        //解析后 得到的msg : BigInteger
        System.out.println("channelRead0 : "+ msg);
        //读次数统计 . 每读一次,计数一次.
        receivedMessages ++;
        //当发送次数 == 读取次数 . 计算结束 .
        if (receivedMessages == FactorialClient.COUNT) {

            // Offer the answer after closing the connection.
            // 将 得到的结果 放入到 answer这个阻塞队列中
            // 然后关闭 channel
            ctx.channel()
                .close()
                .addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) {
                    boolean offered = answer.offer(msg);
                    assert offered;
                }
            });
        }
    }


    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        /**
         * 如果有异常抛出,
         * 打印出异常 , 并关闭channel
         */
        cause.printStackTrace();
        ctx.close();
    }

    /**
     * 发送数据
     */
    private void sendNumbers() {
        /**
         * 不要一定性发送 , 超过4096个数字. 即4096个字节
         */
        // Do not send more than 4096 numbers.
        ChannelFuture future = null;
        //这里我自己做了个修改,想验证一些问题
        // 1. readableBytes != LoggingHandler中输出的数据量
        // 1. 每一个write动作 ,对应一个read动作
        // 1. 每个flush动作 , 对应一个readComplete动作
        for (int i = 0; i < 5 ; i++) {
            System.out.println("sendNumbers : "+next);
            future = ctx.write(Integer.valueOf(next));
            next+=999999999;
        }
//        if (next <= FactorialClient.COUNT) {
        //这里判断 ,如果循环异常退出了
        // 则这里会加入listener ,本次发送结束后, 再次执行sendNumbers
//            assert future != null;
//            future.addListener(numberSender);
//        }
        //刷新缓存
        ctx.flush();
    }

    /**
     * 这里加了个保险
     */
    private final ChannelFutureListener numberSender = new ChannelFutureListener() {
        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
            // 发送成功 , 再次执行发送
            if (future.isSuccess()) {
                sendNumbers();
            } else {
                //发送失败 , 关闭channel
                future.cause().printStackTrace();
                future.channel().close();
            }
        }
    };
}
